package com.rui.study.commons.study.http;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.RequestEntity;
import org.springframework.http.ResponseEntity;
import org.springframework.web.client.RestTemplate;

import java.util.LinkedList;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public abstract class AbstractConcurrentRequest {

    public static final String DEFAULT_AGGREGATION_RESULT = "DEFAULT.AGGREGATION.RESULT";

    private static final Logger LOGGER = LoggerFactory.getLogger(AbstractConcurrentRequest.class);

    /**
     * 默认线程数
     */
    private static final int DEFAULT_THREAD_COUNT = 100;

    private static int number;



    /**************************************************
     *                 INSTANCE FIELD                 *
     **************************************************/

    private final int instanceNumber = number++;

    private RestTemplate restTemplate = new RestTemplate();

    /**
     * 线程数
     */
    private int threadCount;



    /**************************************************
     *                  CONSTRUCTOR                   *
     **************************************************/

    public AbstractConcurrentRequest() {
        this.threadCount = DEFAULT_THREAD_COUNT;
    }

    public AbstractConcurrentRequest(int threadCount) {
        this.threadCount = threadCount;
    }

    public AbstractConcurrentRequest(RestTemplate restTemplate, int threadCount) {
        this.restTemplate = restTemplate;
        this.threadCount = threadCount;
    }



    /**************************************************
     *                 PUBLIC METHOD                  *
     **************************************************/

    /**
     * 当前实例个数编号
     * @return
     */
    public int getInstanceNumber() {
        return instanceNumber;
    }



    /**
     * 并发HTTP请求
     * @param responseType
     * @return
     */
    public AggregationResult concurrencyExchange(final Class<?> responseType) {
        return concurrencyExchange(null, responseType);
    }


        /**
         * 并发HTTP请求
         * @param requestEntity
         * @param responseType
         * @return
         */
    public AggregationResult concurrencyExchange(final RequestEntity requestEntity, final Class<?> responseType) {
        // 聚合结果
        final AggregationResult aggregationResult = this.new AggregationResult();
        int threadCount = this.threadCount;

        final CyclicBarrier cyclicBarrier = new CyclicBarrier(threadCount);
        final CountDownLatch countDownLatch = new CountDownLatch(threadCount);
        final Lock lock = new ReentrantLock();
        int count = 0;

        while (count < threadCount) {
            // 创建线程
            final String threadName = "concurrent-request-" + count++;
            Thread thread = new Thread(new Runnable() {
                public void run() {
                    try {

                        LOGGER.info("Arrive to barrier!");
                        cyclicBarrier.await();
                        LOGGER.info("{} is run...", threadName);


                        RequestEntity customRequestEntity = requestEntity;
                        if (customRequestEntity == null){
                            customRequestEntity = buildRequestEntity();
                        }
                        // HTTP请求
                        ResponseEntity<?> result = restTemplate.exchange(customRequestEntity, responseType);

                        lock.lock();
                        try {
                            // 汇聚结果
                            processAggregationResult(result, threadName, aggregationResult);
                        } finally {
                            lock.unlock();
                        }

                        countDownLatch.countDown();
                        LOGGER.info("{} is over", threadName);
                    } catch (InterruptedException e) {
                        LOGGER.error("Catch an 'InterruptedException', caused by: {}", e.getMessage());
                    } catch (BrokenBarrierException e) {
                        LOGGER.error("Catch an 'BrokenBarrierException', caused by: {}", e.getMessage());
                    }
                }
            }, threadName);
            thread.start();
        }

        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            LOGGER.error("Catch an 'InterruptedException', caused by: {}", e.getMessage());
        }

        LOGGER.info("Concurrent [{}] requests are complete", this.threadCount);
        return aggregationResult;
    }


    protected abstract RequestEntity buildRequestEntity();


    /**************************************************
     *                PROTECTED METHOD                *
     **************************************************/

    /**
     * 单个请求的结果处理，由子类实现。
     * 需要注意，对aggregationResult的操作是线程安全的
     * @param result
     * @param threadName
     * @param aggregationResult
     * @return
     */
    protected void processAggregationResult(ResponseEntity<?> result, String threadName,
                                            final AggregationResult aggregationResult) {
        processListResult(DEFAULT_AGGREGATION_RESULT, result, aggregationResult);
    }

    /**
     * 单个请求的结果处理，由子类实现。
     * 需要注意，对aggregationResult的操作是线程安全的
     * @param attrName
     * @param value
     * @param aggregationResult
     * @return
     */
    protected <T> void processListResult(String attrName, T value,
                                         final AggregationResult aggregationResult) {
        Object obj = aggregationResult.getAttribute(attrName);
        LinkedList<T> r;
        if (obj == null) {
            r = new LinkedList<T>();
        } else {
            r = (LinkedList<T>) obj;
        }
        r.addLast(value);
        aggregationResult.setAttribute(attrName, r);
    }



    /**************************************************
     *                 GETTER/SETTER                  *
     **************************************************/

    public int getThreadCount() {
        return threadCount;
    }

    public void setThreadCount(int threadCount) {
        this.threadCount = threadCount;
    }



    /**************************************************
     *                  INNER CLASS                   *
     **************************************************/

    public class AggregationResult {
        private ConcurrentHashMap attr = new ConcurrentHashMap();

        public Object getAttribute(String attribute) {
            return attr.get(attribute);
        }

        public void setAttribute(String attribute, Object value) {
            this.attr.put(attribute, value);
        }
    }

}


